package com.ziki.test.ftp;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.math.BigDecimal;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ziki.common.util.net.FtpUtil;
import com.ziki.common.util.net.FtpUtil.UploadStatus;
import com.ziki.test.proxool.PoolTest;

public class FtpGatewayThread extends Thread {
    private static final Logger log = LoggerFactory.getLogger(FtpGatewayThread.class);
    // MAP中的Key
    private static final String OTA_DATA_KEY = "otadata";
    private static final String TXT_DATA_KEY = "txtdata";
    // ftp工具
    private FtpUtil ftp;
    // ftp的参数
    /** FTP地址 */
    private String hostname;
    /** FTP用户名 */
    private String username;
    /** FTP密码 */
    private String password;
    /** FTP端口 */
    private int port;
    /** FTP连接失败时,重新连接的次数 */
    private int reconnectCount = 10;
    /** FTP两次重新连接的时间间隔 */
    private int connectInterval = 10000;

    /** FTP服务器当中的OTA数据的存放路径 */
    private String otaPath = "/";
    /** OTA数据文件名的指定前缀 */
    private String otaPrefix = "SMS_SEND_";
    /** FTP服务器当中的文本数据的存放路径 */
    private String txtPath = "/";
    /** 文本数据文件名的指定前缀 */
    private String txtPrefix = "SMS_SEND_";
    /** 上传FTP文件名的时间戳格式 */
    private String timeFormat = "yyyyMMddHHmmssSSS";
    /** 时间格式化类 */
    private SimpleDateFormat datFormat = new SimpleDateFormat(timeFormat);
    /** 上次发送心跳包的时间 */
    private Date activeTime = new Date();
    /** 两次发送心跳包的时间间隔 */
    private int activeInterval = 120000;

    /**
     * 移动的发送号码，如SP号
     */
    private String senderNo = "10086100";

    /**
     * 查询语句
     */
    private String selectSql = "SELECT T.NUMID, T.VC2MOBILE, T.VC2CONTENT, T.ISOTA FROM USR_SMS_SEND T ORDER BY T.DATCREATE FOR UPDATE";

    /** 线程运行标志 */
    private boolean isRun = true;

    FtpGatewayThread(String hostname, int port, String username, String password) {
        this.hostname = hostname;
        this.port = port;
        this.username = username;
        this.password = password;
        this.ftp = new FtpUtil();
    }

    @Override
    public void run() {
        log.info("Ftp线程启动......");
        int connectErrorCount = 0;
        while (isRun) {
            /** FTP连接过程的处理 **/
            try {
                // FTP连接
                if (!ftp.isConnected()) {
                    if (ftp.connect(hostname, port, username, password)) {
                        connectErrorCount = 0;
                    } else {
                        connectErrorCount++;
                    }
                }
            } catch (IOException e) {
                log.error("FTP连接出错", e);
                connectErrorCount++;
            } finally {
                if (!ftp.isConnected()) {
                    // FTP连续连接出错次数超过规定值则停止连接
                    if (connectErrorCount < reconnectCount) {
                        // 未超过则等待nextConncectSleepTime秒后再试
                        sleep(connectInterval);
                        continue;
                    } else {
                        // TODO FTP连接出错处理：停止线程，写入数据库等
                        writeFtpThreadFlag("stop");
                        isRun = false;
                        break;
                    }
                }
            }

            /** 数据取得过程的处理 **/
            Map<String, String> mpData = null;
            try {
                mpData = getData();
            } catch (SQLException e) {
                log.error("数据取得异常", e);
                // 数据取得异常，线程停止，需要人工处理
                writeFtpThreadFlag("stop");
                isRun = false;
                break;
            }

            /** 数据上传过程的处理 **/
            ByteArrayInputStream bis = null;
            try {
                String otaData = mpData.get(OTA_DATA_KEY);
                String txtData = mpData.get(TXT_DATA_KEY);

                // 上传文本文件
                if (txtData != null && txtData.length() > 0) {
                    bis = new ByteArrayInputStream(txtData.getBytes("GBK"));
                    UploadStatus txtStauts = null;
                    try {
                        txtStauts = ftp.upload(bis, getTxtPathAndName());
                    } catch (IOException e) {
                        log.error("上传文本文件失败", e);
                    }
                    if (txtStauts != UploadStatus.Upload_New_File_Success) {
                        log.error("上传文本文件失败:" + txtStauts);
                        // TODO 需要上传失败后的处理
                    }
                    bis.close();
                }

                // 上传配置编码文件
                if (otaData != null && otaData.length() > 0) {
                    bis = new ByteArrayInputStream(otaData.getBytes("GBK"));
                    UploadStatus otaStauts = null;
                    try {
                        otaStauts = ftp.upload(bis, getOtaPathAndName());
                    } catch (IOException e) {
                        log.error("上传OTA文件失败", e);
                    }
                    if (otaStauts != UploadStatus.Upload_New_File_Success) {
                        log.error("上传OTA文件失败:" + otaStauts);
                        // TODO 需要上传失败后的处理
                    }
                    bis.close();
                }
                // 隔一定时间发送心跳命令,防止服务端连接超时
                if (new Date().getTime() - activeTime.getTime() > activeInterval) {
                    ftp.active();
                    activeTime = new Date();
                }
            } catch (IOException e) {
                log.error("FTP异常", e);
            }
            // 程序执行完,Sleep
            sleep(1000);
        }
        log.warn("Ftp线程结束......");
    }

    /**
     * 处理各种标志
     * 
     * @param type
     */
    private void writeFtpThreadFlag(String type) {
        // TODO 处理各种标志
        if ("stop".equalsIgnoreCase(type)) {
            log.error("FTP线程停止");
        }

    }

    /**
     * 取得数据库的数据
     * 
     * @return
     * @throws SQLException
     */
    private Map<String, String> getData() throws SQLException {
        Map<String, String> map = new HashMap<String, String>();
        List<BigDecimal> msgIds = new ArrayList<BigDecimal>();
        StringBuffer sbTxt = new StringBuffer(1000);
        StringBuffer sbOta = new StringBuffer(1000);

        Connection conn = null;
        Statement st = null;
        PreparedStatement pstmt = null;
        ResultSet rs = null;

        try {
            PoolTest pool = PoolTest.getInstance();
            conn = pool.getConnection();
            // 不自动提交
            conn.setAutoCommit(false);
            st = conn.createStatement();
            // 删除下发数据
            pstmt = conn.prepareStatement("DELETE FROM USR_SMS_SEND WHERE NUMID=?");
            // 查询下发数据
            rs = st.executeQuery(selectSql);
            while (rs.next()) {
                int msgType = rs.getInt("ISOTA");
                // 是OTA配置
                if (msgType == 1) {
                    sbOta.append(rs.getString("VC2MOBILE"))
                            .append("|")
                            .append(senderNo)
                            .append("|")
                            .append("VC2CONTENT")
                            .append("\n");
                } else { // 是文本配置
                    sbTxt.append(rs.getString("VC2MOBILE"))
                            .append("|")
                            .append(senderNo)
                            .append("|")
                            .append("VC2CONTENT")
                            .append("\n");
                }
                // 保存数据的ID
                msgIds.add(rs.getBigDecimal("NUMID"));
            }

            // 根据保存的ID删除查询出的数据
            int idCount = 0;
            for (BigDecimal id : msgIds) {
                pstmt.setBigDecimal(1, id);
                pstmt.addBatch();
                idCount++;
                // 添加一定数量批量执行一次
                if (idCount % 200 == 0) {
                    pstmt.executeBatch();
                }
            }
            // 有需要删除的数据时，执行完剩余的批量
            if (idCount > 0)
                pstmt.executeBatch();
            // 提交操作
            conn.commit();
        } catch (SQLException e) {
            try {
                // 异常回滚
                conn.rollback();
            } catch (SQLException e1) {
                log.error("数据库rollback异常", e1);
            }
            log.error("数据库异常", e);
            throw e;
        } finally {
            // 资源关闭
            try {
                if (rs != null)
                    rs.close();
                if (st != null)
                    st.close();
                if (pstmt != null)
                    pstmt.close();
                if (conn != null)
                    conn.close();
            } catch (SQLException e) {
                log.error("数据库close异常", e);
            }
        }
        // 保存取得的数据
        map.put(TXT_DATA_KEY, sbTxt.toString());
        map.put(OTA_DATA_KEY, sbOta.toString());
        return map;
    }

    public String getOtaPathAndName() {
        String date = datFormat.format(new Date());
        return this.otaPath + this.otaPrefix + date;
    }

    public String getTxtPathAndName() {
        String date = datFormat.format(new Date());
        return this.txtPath + this.txtPrefix + date;
    }

    private void sleep(int sleeptime) {
        try {
            Thread.sleep(sleeptime);
        } catch (InterruptedException e) {
            log.warn("sleep error", e);
        }
    }

    public int getReconnectCount() {
        return reconnectCount;
    }

    /**
     * 设置连接FTP的重试次数
     * 
     * @param reconnectCount
     */
    public void setReconnectCount(int reconnectCount) {
        this.reconnectCount = reconnectCount;
    }

    public int getNextConncectSleepTime() {
        return connectInterval;
    }

    /**
     * 设置重新连接FTP的时间
     * 
     * @param nextConncectSleepTime
     */
    public void setNextConncectSleepTime(int nextConncectSleepTime) {
        this.connectInterval = nextConncectSleepTime;
    }

    public String getOtaPath() {
        return otaPath;
    }

    /**
     * 设置OTA文件在FTP上的保存路径
     * 
     * @param otaPath
     */
    public void setOtaPath(String otaPath) {
        this.otaPath = otaPath;
    }

    public String getOtaPrefix() {
        return otaPrefix;
    }

    /**
     * 设置OTA文件的前缀
     * 
     * @param otaPrefix
     */
    public void setOtaPrefix(String otaPrefix) {
        if (otaPrefix.endsWith("/")) {
            this.otaPrefix = otaPrefix;
        } else {
            this.otaPrefix = otaPrefix + "/";
        }
    }

    public String getTxtPath() {
        return txtPath;
    }

    /**
     * 设置文本文件在FTP上的保存路径
     * 
     * @param txtPath
     */
    public void setTxtPath(String txtPath) {
        if (txtPath.endsWith("/")) {
            this.txtPath = txtPath;
        } else {
            this.txtPath = txtPath + "/";
        }
    }

    public String getTxtPrefix() {
        return txtPrefix;
    }

    /**
     * 设置文本文件的前缀
     * 
     * @param txtPrefix
     */
    public void setTxtPrefix(String txtPrefix) {
        this.txtPrefix = txtPrefix;
    }

    public String getTimeFormat() {
        return timeFormat;
    }

    /**
     * 设置上传文件后缀时间戳的格式
     * 
     * @param timeFormat
     */
    public void setTimeFormat(String timeFormat) {
        this.timeFormat = timeFormat;
        this.datFormat.applyPattern(timeFormat);
    }

    public int getActiveInterval() {
        return activeInterval;
    }

    /**
     * 设置两次心跳包的间隔时间（毫秒）
     * 
     * @param activeInterval
     */
    public void setActiveInterval(int activeInterval) {
        this.activeInterval = activeInterval;
    }

    /**
     * 检测线程是否运行
     * 
     * @return
     */
    public boolean isRun() {
        return isRun;
    }

    /**
     * 停止线程
     */
    public void stopTread() {
        this.isRun = false;
    }

    public String getSenderNo() {
        return senderNo;
    }

    /**
     * 设置发送者的号码，类似SPNUMBER
     * 
     * @param senderNo
     */
    public void setSenderNo(String senderNo) {
        this.senderNo = senderNo;
    }

    public String getSelectSql() {
        return selectSql;
    }

    /**
     * 设置查询数据库的SQL
     * 
     * @param selectSql
     */
    public void setSelectSql(String selectSql) {
        this.selectSql = selectSql;
    }
}
